 
/* 
 * Copyright (C) Ignacio Lucas
 */
 
/*
 *  io_handler, hecho Ignacio Lucas
 *
 *  Encapsulamiento de mecanismo de multiplexacion de eventos de Entrada/Salida de red.
 *  epoll() usado como mecanismo de multiplexacion.
 *
 *  Ignacio Lucas <ignacio.lucas@gmail.com>
 */
#include "common.h"
#include "io_handler.h"

// servicio de timers.
#include "timer_service.h"

#include <fcntl.h>
// epoll usado como mecanismo de multiplexacion
#include <sys/epoll.h>

// para la serializacion de llamadas al timer service.
#include "exclusive_call.h"


/** 
	significado de codigo de error EPOLLHUP:
	-----------------

	http://lkml.indiana.edu/hypermail/linux/kernel/0404.0/0470.html
	
	For sockets at least, POLLHUP should indicate the socket is fully closed, so that reading and writing will both fail.
	Thus it makes sense that POLLHUP is not ignorable, although curiously select() only treats 
	POLLHUP as an _input_ condition, so it won't wake something that's waiting only for output readiness. poll() will
	always wake even if you're only waiting for POLLOUT.
**/
#define MAX_EVENTS  1500

// timer usado por default para el epoll. Valor expresado en milisegundos.
#define DEFAULT_TIMER  (int)(2<<30)

typedef struct s_io_handler
{
  struct epoll_event *events;
  int nevents;
  int epfd;
  event_fired_callback callback;

  // sistema de comunicacion via pipes para la actualizacion de timers del epoll.
  int pipe_timer_fds[2];

  // timer actual de timeout para epoll.
  int timer;
	
  // servicio para manejo de timers.	
  timer_service_t timer_service;
	
  // sistema de comunicacion via pipes para la comunicacion entre jobs
  int pipe_job_comm_fds[2];
  
} io_handler;

// wrapper para pasar por el pipe de job_comm un puntero
typedef struct s_io_handler_job_event_wrapper
{
	job_event_t event;
	
} io_handler_job_event_wrapper;

// job_event que se emitira cuando surge un evento de EPOLLIN sobre el PIPE del timer.
static job_event pipe_timer_event;

// job_event que se emitira cuando surge un evento de EPOLLIN sobre el PIPE de comunicacion entre jobs.
static job_event pipe_job_comm_event;

static inline int io_handler_callback_on_timeout(job_event_t jevent, void *caller)
{
	io_handler *ioh = (io_handler *)caller;

	// ponemos la marca de que se salio por timeout.
	jevent->fired_events = JOB_EVENT_TIMEOUT;
	
	// ante un evento de timeout informamos que solo este evento está pendiente.
	return ioh->callback(jevent, 0);

}

static int io_handler_init_pipe(io_handler *ioh, int *fds, const char* pipedef, job_event *jevent)
{
	int ret_pipe = 0;
	
	// cargamos los pipes para configurar las notificaciones de cambios de timers.
	ret_pipe = pipe(fds);
	if (ret_pipe != 0)
	{
		logger_io_error( "[io_handler] Error al inicializar pipe '%s' : %d", pipedef, ret_pipe);
		return ret_pipe;
	}
	
	// configuramos de tipo no bloqueante el fd de lectura.
	if ((ret_pipe = fcntl(fds[0], F_SETFL, O_NONBLOCK)) != 0)
	{
		logger_io_error( "[io_handler] Error seteando no bloqueante el pipe de lectura para '%s' : %d", pipedef, errno);
		return ret_pipe;
	}
	
	// configuramos el job_event del pipe.
	jevent->job = NULL; // en este caso no existe un job asociado.
	jevent->source = JOB_SOURCE_PIPE;
	jevent->data.pipe_conf.pipe_fds = fds;  // Le cargo los 2 fds del pipe.
	jevent->data.pipe_conf.fd_offset = 0;  // Le indico que el fd de interes es el primero del array (fd de lectura del pipe).
	jevent->operation = JOB_EVENT_OP_ADD;
	jevent->asked_events = JOB_EVENT_INPUT; // queremos ser informados del input en el fd de read del pipe.
	
	// solicitamos edge-triggered pues el fd de lectura esta puesto no bloqueante.
	jevent->persistent=1;
	jevent->edge_or_level=0;
	ret_pipe = io_handler_register_event_interest(ioh, jevent);
	
	if (ret_pipe != 0)
	{
	    logger_io_error( "[io_handler] Error inicializando configuracion de pipe '%s' en epoll: %d", pipedef, ret_pipe);
		return ret_pipe;
	}	
	logger_io_debug( "[io_handler] configuracion exitosa de pipe '%s'", pipedef);
	return 0;
}

static int io_handler_init_pipe_timer(io_handler *ioh)
{
	return io_handler_init_pipe(ioh, ioh->pipe_timer_fds, "timers", &(pipe_timer_event));
}

static int io_handler_init_pipe_job_comm(io_handler *ioh)
{
	return io_handler_init_pipe(ioh, ioh->pipe_job_comm_fds, "job comm", &(pipe_job_comm_event));
}

static int io_handler_init_epoll(io_handler *ioh)
{
	// inicializacion del fd del epoll a un valor invalido.
	ioh->epfd = -1;
	
	// inicializacion del timer del epoll a un valor muy alto.
	ioh->timer = DEFAULT_TIMER;
	
	ioh->epfd=epoll_create(ioh->nevents);
	if (ioh->epfd==-1)
	{
		logger_io_error( "[io_handler] error en epoll_create: %s", strerror(errno));
		return 1;
	}
	else
	{
		logger_io_info("%s","[io_handler] epoll_create init ok");
		
		ioh->events = malloc(ioh->nevents * sizeof(struct epoll_event));
		if (ioh->events == NULL) {
			logger_io_error("%s","[io_handler] sin memoria para eventos epoll");
			return 2;
		}
		if (io_handler_init_pipe_timer(ioh) != 0)
		{
			return 2;
		}
		if (io_handler_init_pipe_job_comm(ioh) != 0)
		{
			return 2;
		}
		logger_io_debug("%s","[io_handler] eventos epoll ok");
		return 0;
	}
}

io_handler_t io_handler_init_max_events(unsigned int max_events, event_fired_callback callback)
{
    io_handler *ioh = NULL;
    int ret = 0;
    
    if ((ioh=(io_handler *)malloc(sizeof(io_handler) )) ==  NULL)
  	{
  		logger_io_error("%s","[io_handler] sin memoria para io_handler_t");
			return NULL;  		
  	}
	ioh->nevents = max_events;
	ioh->events = NULL;
	ioh->callback = callback;
	ioh->timer_service = NULL;

	// inicializacion con file descriptors invalidos para los pipes.
	ioh->pipe_timer_fds[0]=ioh->pipe_timer_fds[1]=ioh->pipe_job_comm_fds[0]=ioh->pipe_job_comm_fds[1]=-1;
    
    ret = io_handler_init_epoll(ioh);
    if (ret != 0)
  	{
  		logger_io_error( "[io_handler] error al inicializar io_handler. ret=%d", ret);
  		io_handler_destroy(ioh);
  		return NULL;
  	}

	// inicializamos el timer_service pasandole el file descriptor del pipe asociado a la operacion de escritura.
	// y la funcion de callback que queremos ejecutar cada vez que surja un timeout.
	ioh->timer_service = timer_service_init(ioh->pipe_timer_fds[1], ioh, io_handler_callback_on_timeout);
	if (ioh->timer_service == NULL)
	{
  		logger_io_error("%s","[io_handler] error al inicializar servicio de timers");
		io_handler_destroy(ioh);
  		return NULL;
	}
  	return (io_handler_t)ioh;
}

io_handler_t io_handler_init(event_fired_callback callback)
{
	return io_handler_init_max_events(MAX_EVENTS, callback);
}

static int io_handler_close_pipes(int fds[])
{
	if (fds[0] != -1)
	{
		close(fds[0]);
	}
	if (fds[1] != -1)
	{
		close(fds[1]);
	}	
	return 0;
}

int io_handler_destroy(io_handler_t ioh_t)
{
	io_handler *ioh = (io_handler_t)ioh_t;
	if (ioh->events != NULL)
	{
		free(ioh->events);
		ioh->events=NULL;
	}
	ioh->callback=NULL;

	// cerramos el pipe de timers. Como consecuencia, se elimina del epoll.
	io_handler_close_pipes(ioh->pipe_timer_fds);
	
	// cerramos el pipe de job comm. Como consecuencia, se elimina del epoll.
	io_handler_close_pipes(ioh->pipe_job_comm_fds);
	
	// cerramos el fd del epoll.
	if (ioh->epfd != -1)
	{
		close(ioh->epfd);
	}

	// liberamos el timer_service
	if (ioh->timer_service != NULL)
	{
		timer_service_destroy(ioh->timer_service);
	}

	free(ioh);
	ioh=NULL;
	return 0;
}

/** 
Ver uso de epoll:

http://www.developerweb.net/forum/showthread.php?t=5702
**/
static int io_handler_op_fd(io_handler_t ioh_t, int fd, int epoll_op, int events, job_event_t regevent, int persistent, int level_triggered)
{
	struct epoll_event e;
	io_handler * ioh = (io_handler *)ioh_t;
	int ret = 0;
	
	memset(&e, 0, sizeof(struct epoll_event));
	
	e.events = (level_triggered)?0:EPOLLET;
	if (persistent == 0)
	{
		e.events |= EPOLLONESHOT;
	}
	e.events |= ((events & JOB_EVENT_INPUT) ? EPOLLIN : 0) | ((events & JOB_EVENT_OUTPUT) ? EPOLLOUT : 0) | ((events & JOB_EVENT_ERR) ? EPOLLERR : 0) | ((events & JOB_EVENT_HUP) ? EPOLLHUP : 0);
	// Guardo info en el puntero de user_data. No se pueden guardar fd y user_data porque e.data es un union.
	e.data.ptr = regevent;
	
	errno=0;
	ret = epoll_ctl (ioh->epfd, epoll_op, fd, &e);
	return ret;
}

/** 
	Estas tres funciones deben ejecutarse sin concurrencia por lo que usamos los servicios de jobs controller para
	poder ejecutarlas de forma serializada. 
**/
static inline int io_handler_configure_timeout(io_handler *ioh, job_event_t regevent)
{
	return timer_service_register_timeout(ioh->timer_service, regevent);
}

static inline int io_handler_remove_timeout(io_handler *ioh, job_event_t regevent)
{
	logger_io_info( "[io_handler] remove_timeout. event: %p", regevent);
	return timer_service_remove_timeout(ioh->timer_service, regevent);
}


static inline void io_handler_process_timeouts(io_handler *ioh, void *nullarg)
{
	timer_service_process_timeouts(ioh->timer_service, &(ioh->timer), DEFAULT_TIMER);
}

static inline int io_handler_send_job_comm(io_handler *ioh, job_event_t event)
{
	int ret_write=0, ret = 0;
	io_handler_job_event_wrapper wrapper;
	
	wrapper.event = event;
	
	ret_write = write(ioh->pipe_job_comm_fds[1], (void *)(&wrapper), sizeof(wrapper));
	if (ret_write != (sizeof(wrapper)))
	{
		logger_io_info( "[io_handler] error escribiendo a pipe de job_comm. err: %d", errno);
		ret=1;
	}
	logger_io_info( "[io_handler] escribi %d al pipe de job_comm. puntero: %p", ret_write, event);
	return ret;	
}

int io_handler_register_event_interest(io_handler_t ioh_t, job_event_t regevent)
{
	int realop = 0;
	int op = 0;
	int fd = -1;
	int ret=0;
	int source_requires_op_fd=1;
	int persistent = 0;
	int level_triggered = 0;
	
	persistent=regevent->persistent;
	level_triggered = regevent->edge_or_level;
	op = regevent->operation;
	
	realop = ((op == JOB_EVENT_OP_ADD)?EPOLL_CTL_ADD:0) | ((op == JOB_EVENT_OP_MOD)?EPOLL_CTL_MOD:0) | ((op == JOB_EVENT_OP_DEL)?EPOLL_CTL_DEL:0);

	switch (regevent->source)
	{
		case JOB_SOURCE_SOCKET:
			fd = socket_get_fd(regevent->data.sock_conf.st);
			break;
		case JOB_SOURCE_PIPE:
			fd = regevent->data.pipe_conf.pipe_fds[regevent->data.pipe_conf.fd_offset];
			break;
		case JOB_SOURCE_JOB_COMM:
			// aqui tenemos que enviar al pipe de job_comm el job_event configurado.
			source_requires_op_fd=0;
		
			ret=io_handler_send_job_comm(ioh_t, regevent);
			//logger_io_info( "[io_handler] send_job_comm dio %d", ret);
			break;
		default:
			// warning!
			break;
	}
	
	if (source_requires_op_fd == 1)
	{
		/** Si tenemos alguna configuracion de timeout, la aplicamos ahora. **/
		if (regevent->asked_events & JOB_EVENT_TIMEOUT)
		{
			ret=exclusive_call(io_handler_configure_timeout,(io_handler *)ioh_t, regevent);
		}
		if (ret == 0)
		{
			ret = io_handler_op_fd(ioh_t, fd, realop, regevent->asked_events, regevent, persistent, level_triggered);
		}
	}
	return ret;
}

static inline int io_handler_is_pipe_timer_update_event(io_handler *ioh, job_event_t jevent)
{
	return  (
				(jevent->source == JOB_SOURCE_PIPE) && 
				(jevent->data.pipe_conf.pipe_fds[jevent->data.pipe_conf.fd_offset]==ioh->pipe_timer_fds[0])
			)?1:0;
}

static inline int io_handler_is_pipe_job_comm_event(io_handler *ioh, job_event_t jevent)
{
	return  (
				(jevent->source == JOB_SOURCE_PIPE) && 
				(jevent->data.pipe_conf.pipe_fds[jevent->data.pipe_conf.fd_offset]==ioh->pipe_job_comm_fds[0])
			)?1:0;
}

static inline int io_handler_clear_timer_update_event(io_handler *ioh)
{
	char value;
 
	// lectura hasta vaciar el pipe
	while (read(ioh->pipe_timer_fds[0],&value,1) > 0);
	
	return 0;
}

static inline void io_handler_handle_job_comm(io_handler *ioh, job_event_t job_comm_event)
{
	// se supone que siempre el source sera de tipo job_comm...
	server_worker_job_t fired_job = NULL, parent_job = NULL;
	server_worker_job_comm_config_t parent_job_comm = NULL;
	int parent_status = -1;
	
	//logger_io_info( "[io_handler] [job_comm] source=%d", job_comm_event->source);
	//logger_io_info( "[io_handler] [job_comm] fired_events=%d", job_comm_event->fired_events);
	
	if (job_comm_event->source == JOB_SOURCE_JOB_COMM)
	{
		fired_job = job_comm_event->data.server_worker_job_comm_conf.job;
		
		if (job_comm_event->fired_events == JOB_EVENT_CHILD_FINISHED)
		{
			// asignamos el estado correspondiente al job hijo.
			server_worker_job_set_status(fired_job,SERVER_WORKER_JOB_DONE);
			
			// verificamos si el padre ya esta esperando a este hijo finalizado.
			parent_job = server_worker_job_get_parent(fired_job);
			parent_status = server_worker_job_get_status(parent_job);
			parent_job_comm = server_worker_job_get_job_comm_config(parent_job);
			
			// restamos uno a la espera del padre.
			parent_job_comm->pending_children--;
			
			// se completo la espera de todos los hijos y el padre esta esperando su finalizacion? entonces, rescheduleamos al padre.
			if ((parent_status == SERVER_WORKER_JOB_PARENT_WAITING) && (parent_job_comm->pending_children == 0))
			{
				// dentro del job_comm_event, el job configurado es el padre (pues es el interesado en ser notificado).
				ioh->callback(job_comm_event, 0);
			}			
		}
		else if (job_comm_event->fired_events == JOB_EVENT_PARENT_WAITING)
		{
			// asignamos el estado correspondiente al job.
			server_worker_job_set_status(fired_job,SERVER_WORKER_JOB_PARENT_WAITING);
			
			// analizamos si no queda ningun hijo pendiente.
			parent_job_comm = server_worker_job_get_job_comm_config(fired_job);
			
			if (parent_job_comm->pending_children == 0)
			{
				// dentro del job_comm_event, el job configurado es el padre (pues es el interesado en ser notificado).
				ioh->callback(job_comm_event, 0);
			}
		}
	}
}

static inline int io_handler_process_job_comms(io_handler *ioh)
{
	// aqui debemos vaciar el pipe de lectura de job_comm hasta no encontrar mas elementos.
	job_event_t job_comm_event = NULL;
	ssize_t size_read = 0;
	size_t expected_read = sizeof(io_handler_job_event_wrapper);
	io_handler_job_event_wrapper wrapper;
	
	//logger_io_info("%s","[io_handler] entrando al ciclo de job_comms");
	while ((size_read = (read(ioh->pipe_job_comm_fds[0], &wrapper, expected_read))) == expected_read)
	{
		//logger_io_info( "[io_handler] tengo job_comm en %p!", wrapper.event);
		job_comm_event = wrapper.event;
		
		io_handler_handle_job_comm(ioh,job_comm_event);		
	}	
	//logger_io_info( "[io_handler] sali de job_comm con %d, %s!",size_read,strerror(errno));
	return 0;
}

int io_handler_process_events(io_handler_t ioh_t)
{
	int nfds=0, n=0, ep_event=0, mapped_event=0, ret=0;
	io_handler *ioh = (io_handler *)ioh_t;
	struct epoll_event *fired_events = (ioh->events);
	job_event_t jevent;

	logger_io_debug("[io_handler] epoll_wait. timer: %u", ioh->timer);
	
	nfds = epoll_wait(ioh->epfd, fired_events, ioh->nevents, ioh->timer);
	if (nfds == -1)
	{
		return -1;
	}
  	for(n=0; (ret==0) && (n < nfds); n++)
	{
	  	ep_event = fired_events[n].events;
	  	
	  	// Reset de los eventos a mapear.
	  	mapped_event = 0;
		mapped_event |= (ep_event & EPOLLHUP)?JOB_EVENT_HUP:0;
		mapped_event |= (ep_event & EPOLLERR)?JOB_EVENT_ERR:0;
		mapped_event |= (ep_event & EPOLLIN)?JOB_EVENT_INPUT:0;
		mapped_event |= (ep_event & EPOLLOUT)?JOB_EVENT_OUTPUT:0;

		jevent = (job_event_t)fired_events[n].data.ptr;

		// le cargamos los flags del evento disparado.
		jevent->fired_events = mapped_event;

		// analizamos el tipo de evento generado. El orden es:
		// ver si se trata de comunicacion de pipes (timer / job_comm).
		// else, se trata de un evento de epoll de la aplicacion.
		
		if (io_handler_is_pipe_timer_update_event(ioh,jevent)==1)
		{
			// depuramos solo un byte del pipe.
			logger_io_info("%s","[io_handler] recibo actualizacion de timer");
			io_handler_clear_timer_update_event(ioh);
		}
		else
		{
			if (io_handler_is_pipe_job_comm_event(ioh,jevent)==1)
			{
				// resolvemos las comunicaciones entre jobs.
				io_handler_process_job_comms(ioh);
			}
			else
			{
				// si teniamos configurado timers sobre este evento, lo quitamos.
				if (jevent->asked_events & JOB_EVENT_TIMEOUT)
				{
					exclusive_call(io_handler_remove_timeout,(io_handler *)ioh_t, jevent);
				}
				
				// llamamos a la funcion de callback que consume los eventos generados.
				ret=ioh->callback(jevent, nfds - n - 1);
			}
		}
	}

	// revisamos actualizaciones de timers.
	exclusive_call(io_handler_process_timeouts,ioh,NULL);
	return ret;
}

